#include "svr_skeleton.h"

extern int worker_add_msg(int worker_idx, char *from,
        unsigned int syn, char *request);

static struct
{/*{{{*/
    MSGQ_RECV msgqrv;
    MSGQ_SEND msgqsd;
    char *msg;
}/*}}}*/
listener_ctl;

static int _respond(char *toqueue, unsigned int syn, char *errcode)
{/*{{{*/
    static int TIMEOUT=1*1000;
    char *respond=listener_ctl.msg;
    int res;

    CLR_BUF(respond);
    para_put_item(respond, errcode);
    int respond_size=strlen(respond);
    res=msgq_send(listener_ctl.msgqsd, NULL, toqueue, syn,
            (unsigned char *)respond, respond_size, TIMEOUT);
    if(res)
    {
        errlog("%s %s: respond to agent fail, res:%d, BAD",
                __FILE__, __func__, res);
    }
    return(0);
}/*}}}*/

static int _usr(char *from, unsigned int syn, char *request)
{/*{{{*/
    static int MAX_NUM=100;
    int idle_num=0;
    int idle[cnfg_info.start];
    int min_busy_client_num=MAX_NUM;
    int min_busy_idx=-1;
    int worker_idx;

    pthread_mutex_lock(&stat_info._worker_.lock);

    for(worker_idx=0; worker_idx<cnfg_info.start; ++worker_idx)
    {
        struct _worker_info *worker=
            &stat_info._worker_.worker[worker_idx];
        if(worker->client_num == 0)
        {   idle[idle_num++]=worker_idx;   }
        else
        {
            if(min_busy_client_num > worker->client_num)
            {
                min_busy_client_num=worker->client_num;
                min_busy_idx=worker_idx;
            }
        }
    }
    if(idle_num > 0) // 选取无连接的worker
    {
        worker_idx=idle[random_int(0, idle_num-1, NULL)];
        errlog("%s: choose idle worker(%d)",
                __func__, worker_idx);
    }
    else if(min_busy_client_num < MAX_NUM) // 选取连接最少的worker
    {
        worker_idx=min_busy_idx;
    }
    else
    {
        errlog("%s: worker full, BAD", __func__);
        pthread_mutex_unlock(&stat_info._worker_.lock);
        _respond(from, syn, SOAR_ERR_CONN_FULL);
        return(-1);
    }
    ++stat_info._worker_.worker[worker_idx].client_num;

    pthread_mutex_unlock(&stat_info._worker_.lock);

    int res=worker_add_msg(worker_idx, from, syn, request);
    if(res)
    {
        errlog("%s: listener send msg to worker(%d) fail, res:%d, BAD",
                __func__, worker_idx, res);
        _respond(from, syn, SOAR_ERR_RESOURCE);
        return(-1);
    }

    return(0);
}/*}}}*/

static void _listener()
{/*{{{*/
    unsigned int quit=IS_FALSE;
    int res;
    
    char *request=listener_ctl.msg;
    int request_size;
    char from[NAME_SIZE+1];
    unsigned int syn;
    static int TIMEOUT=30*1000;
    char *curr;
    char *level;
    while(!quit)
    {
        res=msgq_recv(listener_ctl.msgqrv, from,
                &syn, (unsigned char *)request, &request_size, TIMEOUT);
        if(res)
        {
            if(res != RET_ERR_TIMEOUT)
            {
                errlog("%s: msgq_recv fail, res:%s, BAD",
                        __func__, res);
            }
            continue;
        }
        if(BUF_IS_CLR(from))
        {
            errlog("%s: cannot determine from, BAD", __func__);
            continue;
        }
        request[request_size]=ZERO;

        // 解析服务
        curr=NULL;
        level=para_get_item(request, &curr);
        if(!level)
        {
            errlog("%s: msg level null, BAD", __func__);
            continue;
        }
        if(strcmp(level, "SYS") == 0)
        {
            char *cmd=para_get_item(request, &curr);
            if(strcmp(cmd, "DOWNSVR") == 0)
            {
                _respond(from, syn, SOAR_ERR_OK);
                quit=IS_TRUE;
            }
        }
        else if(strcmp(level, "USR") == 0)
        {
            _usr(from, syn, curr);
        }
        else
        {
            errlog("%s: msg level(%s) invalid, BAD",
                    __func__, level);
            _respond(from, syn, SOAR_ERR_MSG);
        }
    }

    errlog("%s: listener shutdown", __func__);
}/*}}}*/

int listener_start()
{/*{{{*/
    char buf[200];
    int res;

    // 消息队列
    char svr_queue[NAME_SIZE+1]; 
    sprintf(svr_queue, "se_svr_%s_%d", SERVER_NAME, cnfg_info.ipckey);
    res=msgq_clear(svr_queue);
    if(res)
    {
        sprintf(buf, "%s: clear queue(%s) fail, res:%d",
                SERVER_NAME, svr_queue, res);
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }
    res=msgq_recv_attach(svr_queue, &listener_ctl.msgqrv);
    if(res)
    {
        sprintf(buf, "%s: attach recv queue(%s) fail, res:%d",
                SERVER_NAME, svr_queue, res);
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }
    res=msgq_send_attach(&listener_ctl.msgqsd);
    if(res)
    {
        sprintf(buf, "%s: attach send queue fail, res:%d",
                SERVER_NAME, res);
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }
    sprintf(buf, "%s: queue ready", SERVER_NAME);
    fprintf(stderr, "%s\n", buf); errlog(buf);

    ALLOC_MULTI_DOERR(listener_ctl.msg, char, MSGQ_MSGSIZE_MAX,
            sprintf(buf, "%s: listener alloc msg fail", SERVER_NAME);
            fprintf(stderr, "%s\n", buf); errlog(buf);
            return(-1);
            );

    /*--- thread listener ---*/
    sprintf(buf, "%s: start listener succeed", SERVER_NAME);
    fprintf(stderr, "%s\n", buf); errlog(buf);
    _listener();

    return(0);
}/*}}}*/
